{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 协程（yield from）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 语法"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# yield from 简要实现\n",
    "_i = iter(EXPR)  # __iter__ -> __await__\n",
    "try:\n",
    "    _y = _i.send(None)  # prime\n",
    "except StopIteration as _e:\n",
    "    _r = _e.value  # 直接就结束了，一次 yield 都没遇上\n",
    "else:\n",
    "    while 1:  # 不遇到 StopIteration 不算完\n",
    "        try:\n",
    "            _s = yield _y  # 照原样 yield 出去，并接受 send 传入的值\n",
    "        except GeneratorExit as _e:  # 处理 close\n",
    "            _i.close()\n",
    "            raise _e\n",
    "        except BaseException as _e:  # 处理其它异常\n",
    "            _x = sys.exc_info()\n",
    "            try:\n",
    "                _y = _i.throw(*_x)\n",
    "            except StopIteration as _e:\n",
    "                _r = _e.value\n",
    "                break\n",
    "        else:\n",
    "            try:\n",
    "                _y = _i.send(_s)  # 接受到的值原样再 send 下去\n",
    "            except StopIteration as _e:\n",
    "                _r = _e.value\n",
    "                break\n",
    "RESULT = _r  # StopIteration 出来的值就是结果"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 已知总是会 send(None)，继续简化\n",
    "_i = iter(EXPR)  # __iter__\n",
    "while 1:  # 不遇到 StopIteration 不算完\n",
    "    try:\n",
    "        _y = _i.send(None)  # 总是 None，也就无所谓 prime\n",
    "    except StopIteration as _e:\n",
    "        _r = _e.value\n",
    "        break\n",
    "    else:\n",
    "        try:\n",
    "            yield _y  # 照原样 yield 出去，不再接受 send 传入的值，因为总是 None\n",
    "        except GeneratorExit as _e:  # 处理 close\n",
    "            _i.close()\n",
    "            raise _e\n",
    "        except BaseException as _e:  # 处理其它异常\n",
    "            _x = sys.exc_info()\n",
    "            try:\n",
    "                _y = _i.throw(*_x)\n",
    "            except StopIteration as _e:\n",
    "                _r = _e.value\n",
    "                break\n",
    "RESULT = _r  # StopIteration 出来的值是结果"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 再去掉异常处理部分代码\n",
    "_i = iter(EXPR)  # __iter__\n",
    "while 1:  # 不遇到 StopIteration 不算完\n",
    "    try:\n",
    "        _y = _i.send(None)  # 总是 None，也就无所谓 prime\n",
    "    except StopIteration as _e:\n",
    "        _r = _e.value\n",
    "        break\n",
    "    else:\n",
    "        yield _y  # 照原样 yield 出去，不再接受 send 传入的值，因为总是 None\n",
    "RESULT = _r  # StopIteration 出来的值就是结果"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 阶段1（定义一个任务）\n",
    "- 一个同步模式的简单任务"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "begin task\n",
      " begin big_step:\n",
      "  begin small_step:\n",
      "    努力工作中...\n",
      "  end small_step with 123\n",
      " end big_step with 123\n",
      "end task\n"
     ]
    }
   ],
   "source": [
    "def one_task():\n",
    "    print(f'begin task')\n",
    "    ... # 其它步骤\n",
    "    print(f' begin big_step:')\n",
    "    \n",
    "    big_result = big_step()  # <---\n",
    "    \n",
    "    print(f' end big_step with {big_result}')\n",
    "    ... # 其它步骤\n",
    "    \n",
    "    print(f'end task')\n",
    "\n",
    "def big_step():\n",
    "    ... # 其它小步骤\n",
    "    print(f'  begin small_step:')\n",
    "    \n",
    "    small_result = small_step()  # <---\n",
    "    \n",
    "    print(f'  end small_step with {small_result}')\n",
    "    ... # 其它小步骤\n",
    "    return small_result  # 123\n",
    "\n",
    "def small_step():\n",
    "    print('    努力工作中...')\n",
    "    return 123  # 完成了\n",
    "\n",
    "\n",
    "# 执行任务\n",
    "one_task()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 阻塞"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "begin task\n",
      " begin big_step:\n",
      "  begin small_step:\n",
      "      躺平中\n",
      "    努力工作中...\n",
      "  end small_step with 123\n",
      " end big_step with 123\n",
      "end task\n"
     ]
    }
   ],
   "source": [
    "from time import sleep\n",
    "\n",
    "def small_step():\n",
    "    print('      躺平中')\n",
    "    sleep(2)  # 阻塞\n",
    "    print('    努力工作中...')\n",
    "    return 123  # 完成了\n",
    "\n",
    "one_task()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 使用yield 使得协程不阻塞"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "begin task\n",
      " begin big_step:\n",
      "  begin small_step:\n",
      "  end small_step with <generator object small_step at 0x000001DCDA679690>\n",
      " end big_step with <generator object small_step at 0x000001DCDA679690>\n",
      "end task\n"
     ]
    }
   ],
   "source": [
    "from time import sleep\n",
    "\n",
    "def small_step():\n",
    "    print('      躺平中')\n",
    "    yield sleep(2)  # 阻塞\n",
    "    print('    努力工作中...')\n",
    "    return 123  # 完成了\n",
    "\n",
    "one_task()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "begin task\n",
      " begin big_step:\n",
      "  begin small_step:\n",
      "      躺平中\n",
      "    努力工作中...\n",
      "  end small_step with 123\n",
      " end big_step with 123\n",
      "end task\n"
     ]
    }
   ],
   "source": [
    "def big_step():\n",
    "    ... # 其它小步骤\n",
    "    print(f'  begin small_step:')\n",
    "    \n",
    "    small_coro = small_step()  # <---\n",
    "    while True:\n",
    "        try:\n",
    "            x = small_coro.send(None)\n",
    "        except StopIteration as e:\n",
    "            small_result = e.value\n",
    "            break\n",
    "        else:\n",
    "            pass\n",
    "    \n",
    "    print(f'  end small_step with {small_result}')\n",
    "    ... # 其它小步骤\n",
    "    return small_result  # 123\n",
    "\n",
    "one_task()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 将阻塞由下游传到上游"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "begin task\n",
      " begin big_step:\n",
      "  begin small_step:\n",
      "      躺平中\n",
      "    努力工作中...\n",
      "  end small_step with 123\n",
      " end big_step with 123\n",
      "end task\n"
     ]
    }
   ],
   "source": [
    "from time import sleep\n",
    "\n",
    "def small_step():\n",
    "    print('      躺平中')\n",
    "    yield (sleep, 2)  # 阻塞\n",
    "    print('    努力工作中...')\n",
    "    return 123  # 完成了\n",
    "\n",
    "one_task()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "begin task\n",
      " begin big_step:\n",
      "  begin small_step:\n",
      "      躺平中\n"
     ]
    },
    {
     "ename": "AssertionError",
     "evalue": "阻塞时间不足",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mAssertionError\u001b[0m                            Traceback (most recent call last)",
      "Cell \u001b[1;32mIn[6], line 12\u001b[0m\n\u001b[0;32m      9\u001b[0m     \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124m'\u001b[39m\u001b[38;5;124m    努力工作中...\u001b[39m\u001b[38;5;124m'\u001b[39m)\n\u001b[0;32m     10\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;241m123\u001b[39m  \u001b[38;5;66;03m# 完成了\u001b[39;00m\n\u001b[1;32m---> 12\u001b[0m \u001b[43mone_task\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
      "Cell \u001b[1;32mIn[1], line 6\u001b[0m, in \u001b[0;36mone_task\u001b[1;34m()\u001b[0m\n\u001b[0;32m      3\u001b[0m \u001b[38;5;241m.\u001b[39m\u001b[38;5;241m.\u001b[39m\u001b[38;5;241m.\u001b[39m \u001b[38;5;66;03m# 其它步骤\u001b[39;00m\n\u001b[0;32m      4\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124mf\u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124m begin big_step:\u001b[39m\u001b[38;5;124m'\u001b[39m)\n\u001b[1;32m----> 6\u001b[0m big_result \u001b[38;5;241m=\u001b[39m \u001b[43mbig_step\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m  \u001b[38;5;66;03m# <---\u001b[39;00m\n\u001b[0;32m      8\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124mf\u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124m end big_step with \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mbig_result\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m'\u001b[39m)\n\u001b[0;32m      9\u001b[0m \u001b[38;5;241m.\u001b[39m\u001b[38;5;241m.\u001b[39m\u001b[38;5;241m.\u001b[39m \u001b[38;5;66;03m# 其它步骤\u001b[39;00m\n",
      "Cell \u001b[1;32mIn[4], line 8\u001b[0m, in \u001b[0;36mbig_step\u001b[1;34m()\u001b[0m\n\u001b[0;32m      6\u001b[0m \u001b[38;5;28;01mwhile\u001b[39;00m \u001b[38;5;28;01mTrue\u001b[39;00m:\n\u001b[0;32m      7\u001b[0m     \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[1;32m----> 8\u001b[0m         x \u001b[38;5;241m=\u001b[39m \u001b[43msmall_coro\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43msend\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;28;43;01mNone\u001b[39;49;00m\u001b[43m)\u001b[49m\n\u001b[0;32m      9\u001b[0m     \u001b[38;5;28;01mexcept\u001b[39;00m \u001b[38;5;167;01mStopIteration\u001b[39;00m \u001b[38;5;28;01mas\u001b[39;00m e:\n\u001b[0;32m     10\u001b[0m         small_result \u001b[38;5;241m=\u001b[39m e\u001b[38;5;241m.\u001b[39mvalue\n",
      "Cell \u001b[1;32mIn[6], line 8\u001b[0m, in \u001b[0;36msmall_step\u001b[1;34m()\u001b[0m\n\u001b[0;32m      6\u001b[0m t1 \u001b[38;5;241m=\u001b[39m time\u001b[38;5;241m.\u001b[39mtime()\n\u001b[0;32m      7\u001b[0m \u001b[38;5;28;01myield\u001b[39;00m (sleep, \u001b[38;5;241m2\u001b[39m)  \u001b[38;5;66;03m# 阻塞\u001b[39;00m\n\u001b[1;32m----> 8\u001b[0m \u001b[38;5;28;01massert\u001b[39;00m time\u001b[38;5;241m.\u001b[39mtime() \u001b[38;5;241m-\u001b[39m t1 \u001b[38;5;241m>\u001b[39m\u001b[38;5;241m=\u001b[39m \u001b[38;5;241m2\u001b[39m, \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m阻塞时间不足\u001b[39m\u001b[38;5;124m\"\u001b[39m\n\u001b[0;32m      9\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124m'\u001b[39m\u001b[38;5;124m    努力工作中...\u001b[39m\u001b[38;5;124m'\u001b[39m)\n\u001b[0;32m     10\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;241m123\u001b[39m\n",
      "\u001b[1;31mAssertionError\u001b[0m: 阻塞时间不足"
     ]
    }
   ],
   "source": [
    "import time\n",
    "from time import sleep\n",
    "\n",
    "def small_step():\n",
    "    print('      躺平中')\n",
    "    t1 = time.time()\n",
    "    yield (sleep, 2)  # 阻塞\n",
    "    assert time.time() - t1 >= 2, \"阻塞时间不足\"\n",
    "    print('    努力工作中...')\n",
    "    return 123  # 完成了\n",
    "\n",
    "one_task()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "begin task\n",
      " begin big_step:\n",
      " end big_step with <generator object big_step at 0x000001DCDBB60190>\n",
      "end task\n"
     ]
    }
   ],
   "source": [
    "def big_step():\n",
    "    ... # 其它小步骤\n",
    "    print(f'  begin small_step:')\n",
    "    \n",
    "    small_coro = small_step()  # <---\n",
    "    while True:\n",
    "        try:\n",
    "            x = small_coro.send(None)\n",
    "        except StopIteration as e:\n",
    "            small_result = e.value\n",
    "            break\n",
    "        else:\n",
    "            yield x  # 将阻塞传到上游\n",
    "    \n",
    "    print(f'  end small_step with {small_result}')\n",
    "    ... # 其它小步骤\n",
    "    return small_result  # 123\n",
    "\n",
    "one_task()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "begin task\n",
      " begin big_step:\n",
      "  begin small_step:\n",
      "      躺平中\n",
      "    努力工作中...\n",
      "  end small_step with 123\n",
      " end big_step with 123\n",
      "end task\n"
     ]
    }
   ],
   "source": [
    "def one_task():\n",
    "    \"\"\"一个任务\"\"\"\n",
    "    print('begin task')\n",
    "    ... # 其它步骤\n",
    "    print(f' begin big_step:')\n",
    "\n",
    "    # big_result = big_step()\n",
    "    big_coro = big_step()\n",
    "    while True:\n",
    "        try:\n",
    "            x = big_coro.send(None)\n",
    "        except StopIteration as e:\n",
    "            big_result = e.value\n",
    "            break\n",
    "        else:\n",
    "            func, arg = x\n",
    "            func(arg)\n",
    "\n",
    "    print(f' end big_step with {big_result}')\n",
    "    ... # 其它步骤\n",
    "\n",
    "    print('end task')\n",
    "\n",
    "one_task()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 总结\n",
    "    - 协程本身并不能消除阻塞\n",
    "    - 协程具有“传染性”\n",
    "    - 协程通过yield 把阻塞换个方式传递给了上游\n",
    "    - 最终阻塞仍然需要被解决"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 阶段2（yield from）"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "begin task\n",
      " begin big_step:\n",
      "  begin small_step:\n",
      "      躺平中\n",
      "    努力工作中...\n",
      "  end small_step with 123\n",
      " end big_step with 123\n",
      "end task\n"
     ]
    }
   ],
   "source": [
    "def big_step():\n",
    "    ... # 其它小步骤\n",
    "    print(f'  begin small_step:')\n",
    "    \n",
    "    small_result = yield from small_step()\n",
    "    # small_coro = small_step()  # <---\n",
    "    # while True:\n",
    "    #     try:\n",
    "    #         x = small_coro.send(None)\n",
    "    #     except StopIteration as e:\n",
    "    #         small_result = e.value\n",
    "    #         break\n",
    "    #     else:\n",
    "    #         pass\n",
    "    \n",
    "    print(f'  end small_step with {small_result}')\n",
    "    ... # 其它小步骤\n",
    "    return small_result  # 123\n",
    "\n",
    "one_task()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "将最末端的遇到阻塞而不得不主动 yield 的协程称之为「主动协程」\n",
    "中间接受到下游的传导而不得不跟随着 yield 的协程称之为「被动协程」\n",
    "\n",
    "- 「主动协程」是最先出栈的位置\n",
    "- 「被动协程」可能有很多层\n",
    "- yield from 大大简化了「被动协程」的编码\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "class YielfFromAble:\n",
    "    def __init__(self, obj):\n",
    "        self.value = obj\n",
    "    \n",
    "    def __iter__(self):\n",
    "        yield self.value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "begin task\n",
      " begin big_step:\n",
      "  begin small_step:\n",
      "    休息一下，马上回来\n",
      "    努力工作中...\n",
      "  end small_step with 123\n",
      " end big_step with 123\n",
      "end task\n"
     ]
    }
   ],
   "source": [
    "import time\n",
    "from time import sleep\n",
    "\n",
    "def small_step():\n",
    "    print('    休息一下，马上回来')\n",
    "    t1 = time.time()\n",
    "    \n",
    "    yield from YielfFromAble((sleep, 2))\n",
    "    \n",
    "    assert time.time() - t1, '睡眠时间不足！'\n",
    "    print('    努力工作中...')\n",
    "    return 123  # 完成了...\n",
    "\n",
    "one_task()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 阶段3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 任务彻底协程化"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<generator object one_task at 0x000001DCDBB61770>"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def one_task():\n",
    "    \"\"\"一个任务\"\"\"\n",
    "    print('begin task')\n",
    "    ... # 其它步骤\n",
    "    print(f' begin big_step:')\n",
    "\n",
    "    # big_result = big_step()\n",
    "\n",
    "    big_result = yield from big_step()\n",
    "\n",
    "    # big_coro = big_step()\n",
    "    # while True:\n",
    "    #     try:\n",
    "    #         x = big_coro.send(None)\n",
    "    #     except StopIteration as e:\n",
    "    #         big_result = e.value\n",
    "    #         break\n",
    "    #     else:\n",
    "    #         func, arg = x\n",
    "    #         func(arg)\n",
    "\n",
    "    print(f' end big_step with {big_result}')\n",
    "    ... # 其它步骤\n",
    "\n",
    "    print('end task')\n",
    "\n",
    "one_task()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 一个通用的任务驱动器"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "class YielfFromAble:\n",
    "    def __init__(self, obj):\n",
    "        self.value = obj\n",
    "    \n",
    "    def __iter__(self):\n",
    "        yield self"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Task:\n",
    "    def __init__(self, coro):\n",
    "        self.coro = coro  # 保存协程对象\n",
    "\n",
    "    def run(self):\n",
    "        print('任务开始运行')\n",
    "        \"\"\"驱动任务运行直到完成\"\"\"\n",
    "        \n",
    "        while True:\n",
    "            try:\n",
    "                x = self.coro.send(None)\n",
    "            except StopIteration as e:\n",
    "                result =  e.value  # 返回最终结果\n",
    "                break\n",
    "            else:\n",
    "                assert isinstance(x, YielfFromAble)\n",
    "                func, arg = x.value\n",
    "                func(arg)\n",
    "        print('任务运行结束')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "任务开始运行\n",
      "begin task\n",
      " begin big_step:\n",
      "  begin small_step:\n",
      "    休息一下，马上回来\n",
      "    努力工作中...\n",
      "  end small_step with 123\n",
      " end big_step with 123\n",
      "end task\n",
      "任务运行结束\n"
     ]
    }
   ],
   "source": [
    "t = Task(one_task())\n",
    "t.run()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "base",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
